package com.farmer.reliablemessage.core;

import com.farmer.reliablemessage.CommonConfiguration;
import com.farmer.reliablemessage.entity.ReliableMessage;
import com.farmer.reliablemessage.event.*;
import com.farmer.reliablemessage.mapper.MessageManagerMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

/**
 * @author tech-famer 2019/12/26 15:06
 * 描述
 */
@Component
public class ReliableMessageManager implements RabbitTemplate.ReturnCallback, RabbitTemplate.ConfirmCallback {

    private static final Logger log = LoggerFactory.getLogger(ReliableMessageManager.class);
    private static int LOAD_LIMIT = 10;
    private MessageManagerMapper messageManagerMapper;
    private ApplicationEventPublisher applicationEventPublisher;
    private RabbitTemplate rabbitTemplate;

    public ReliableMessageManager(MessageManagerMapper messageManagerMapper, ApplicationEventPublisher applicationEventPublisher, RabbitTemplate rabbitTemplate, CachingConnectionFactory connectionFactory) {
        this.messageManagerMapper = messageManagerMapper;
        this.applicationEventPublisher = applicationEventPublisher;
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(this);
        rabbitTemplate.setConfirmCallback(this);
        this.rabbitTemplate = rabbitTemplate;
        log.info("SendMessageThread load success!");
    }

    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            this.messageManagerMapper.updateMessageToOK(correlationData.getId());
            log.debug("reliablemessage[{}] send success!", correlationData.getId());
        } else {
            this.messageManagerMapper.updateFailTimes(correlationData.getId(), cause);
            log.error("reliablemessage[{}] send failed, cause: {}", correlationData.getId(), cause);
        }

    }

    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        String correlationId = message.getMessageProperties().getCorrelationIdString();
        this.messageManagerMapper.updateFailTimes(correlationId, replyText);
        log.error("reliablemessage[{}] send failed, return code:[{}], cause:[{}], exchange:[{}], routingkey:[{}]", correlationId, replyCode, replyText, exchange, routingKey);
    }

    @Async
    @EventListener(
            classes = {ReliableMessageSendEvent.class}
    )
    public void onSendEvent(ReliableMessageSendEvent event) {
        log.debug("received a new reliablemessage send event...");
        ReliableMessage message =  event.getMessage();
        this.rabbitTemplate.convertAndSend(message.getExchange(), message.getRoutingkey(), message.getPayload(), new CorrelationData(message.getMsgid()));
    }

    @Async
    @EventListener(
            classes = {ReliableMessageResendEvent.class}
    )
    public void onResendEvent(ReliableMessageResendEvent event) {
        log.debug("received batch reliablemessage send event...");
        List<ReliableMessage> messages = event.getMessages();
        if (messages != null) {
            for (ReliableMessage message : messages) {
                this.rabbitTemplate.convertAndSend(message.getExchange(), message.getRoutingkey(), message.getPayload(), new CorrelationData(message.getMsgid()));
            }

            if (messages.size() >= LOAD_LIMIT) {
                this.applicationEventPublisher.publishEvent(new ReliableMessageSendContinueEvent(this));
            }
        }
    }

    @Async
    @EventListener(
            classes = {ReliableMessageSendOkEvent.class}
    )
    @Transactional(
            rollbackFor = {Exception.class}
    )
    public void onSendOkEvent(ReliableMessageSendOkEvent event) {
        log.debug("received reliablemessage send ok event...");

        try {
            this.messageManagerMapper.moveMessageToHistory();
            this.messageManagerMapper.deleteSucceedMessage();
        } catch (Exception var3) {
            log.error("MoveMessageToHistoryThread move succeed message failed!, {}", var3);
            throw var3;
        }
    }

    @Async
    @EventListener(
            classes = {ReliableMessageSendFailEvent.class}
    )
    @Transactional(
            rollbackFor = {Exception.class}
    )
    public void onSendFailEvent(ReliableMessageSendFailEvent event) {
        log.debug("received reliablemessage send fail event...");

        try {
            this.messageManagerMapper.moveMessageToFailed();
            this.messageManagerMapper.deleteOverFailedTimesMessage();
        } catch (Exception var3) {
            log.error("MoveFailedMessageThread move failed message failed!, {}", var3);
            throw var3;
        }
    }

    @Async
    @EventListener(
            classes = {ReliableMessageSendContinueEvent.class}
    )
    public void onSendContinueEvent(ReliableMessageSendContinueEvent event) {
        log.debug("received reliablemessage send continue event...");

        try {
            String lockID = CommonConfiguration.getLockID();
            this.messageManagerMapper.lockMessage(lockID);
            List<ReliableMessage> messageList = this.messageManagerMapper.getMessages(lockID);
            if (messageList != null && !messageList.isEmpty()) {
                this.applicationEventPublisher.publishEvent(new ReliableMessageResendEvent(this, messageList));
            }
        } catch (Exception var5) {
            log.error("LockMessageThread lock reliablemessage failed, {}", var5);
        }

    }
}

